package com.lagou.hbase.hw;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

public class BuildTable {

    public static final String TABLE_NAME = "testrelation";
    public static final String CF_NAME = "friends";

    public static void main(String[] args) throws Exception {
        // 获取连接
        Connection conn = createConnection();

        // 创建用户关系表
        ArrayList<String> cfNames = new ArrayList<>(1);
        cfNames.add(CF_NAME);
        createTable(conn, TABLE_NAME, cfNames);

        // 插入模拟数据
        generateMockData(conn);

        // 查看插入结果
        scanTable(conn, TABLE_NAME);

    }

    /**
     * 创建新连接
     *
     * @return
     * @throws IOException
     */
    private static Connection createConnection() throws IOException {
        // 获取一个配置文件对象
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "com01,com02,com03");
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        // 通过conf获取到hbase集群的连接
        return ConnectionFactory.createConnection(conf);
    }

    /**
     * 创建表
     *
     * @param conn
     * @param tableName
     * @param cfNames
     * @throws IOException
     */
    private static void createTable(Connection conn, String tableName, List<String> cfNames) throws IOException {
        HBaseAdmin admin = (HBaseAdmin) conn.getAdmin();
        // 创建Htabledesc描述器，表描述器
        final HTableDescriptor worker = new HTableDescriptor(TableName.valueOf(tableName));
        for (String cfName : cfNames) {
            // 指定列族
            worker.addFamily(new HColumnDescriptor(cfName));
        }

        admin.createTable(worker);

        System.out.println(tableName + "表创建成功！");
    }

    /**
     * 插入模拟数据
     *
     * @param conn
     * @throws IOException
     */
    private static void generateMockData(Connection conn) throws IOException {
        // 获取table对象
        final Table worker = conn.getTable(TableName.valueOf(TABLE_NAME));
        final String prefix = "uid";

        final List<Put> allMockData = new ArrayList<>();

        for (int i = 0; i < 10; i++) {
            // 生成当前用户rowkey
            String masterRowkey = prefix.concat(String.valueOf(i));

            for (int j = 0; j < 10; j++) {
                // 如果 i=j，则说明是自己和自己，那么自己和自己不能成为好友，跳过
                if (i == j) {
                    continue;
                }

                // 生成当前用户的朋友rowkey
                String friendRowkey = prefix.concat(String.valueOf(j));

                // 准备put对象，rowkey 即为主人id，即 masterRowkey
                Put put = new Put(Bytes.toBytes(masterRowkey));
                // 列名和列值都为好友id，即 friendRowkey
                put.addColumn(Bytes.toBytes(CF_NAME), Bytes.toBytes(friendRowkey), Bytes.toBytes(friendRowkey));

                // 加入待插入列表
                allMockData.add(put);
            }

        }

        // 批量插入所有模拟数据
        worker.put(allMockData);

        //关闭table对象
        worker.close();
        System.out.println("模拟数据已成功插入到" + TABLE_NAME + "表！");
    }


    /**
     * 扫描全表
     *
     * @param conn
     * @param tableName
     * @throws IOException
     */
    private static void scanTable(Connection conn, String tableName) throws IOException {
        System.out.println("全表扫描" + tableName + "表结果如下：");

        //准备table对象
        final Table worker = conn.getTable(TableName.valueOf(tableName));
        //准备scan对象
        final Scan scan = new Scan();

        //执行扫描
        final ResultScanner resultScanner = worker.getScanner(scan);
        for (Result result : resultScanner) {
            //获取到result中所有cell对象
            final Cell[] cells = result.rawCells();

            //遍历打印
            for (Cell cell : cells) {
                final String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                final String f = Bytes.toString(CellUtil.cloneFamily(cell));
                final String column = Bytes.toString(CellUtil.cloneQualifier(cell));
                final String value = Bytes.toString(CellUtil.cloneValue(cell));
                System.out.println("rowkey-->" + rowkey + "--;cf-->" + f + ";column-->" + column + "--;value-->" + value);
            }
        }

        worker.close();
    }

}
